package com.yanggu.flink.datastream_api.transform

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

/**
 * map方法, 每来一条数据就会调用一次, 将一种数据类型转换成另外一种数据类型
 * 一般用作数据类型的转换, 常见如从Kafka读取String的数据, 然后使用map转换成样例类
 */
object MapFunctionDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.fromCollection(Seq(1, 2, 3))
      //可以传入匿名函数
      .map(data => data * 2)
      //或者自定义类实现MapFunction接口, 重写map方法
      .map(new MyMapFunction2)
      .print("map test>>>").setParallelism(1)

    env.execute()
  }

}

/**
 *
 * @tparam I 输入的数据类型
 * @tparam O 输出的数据类型
 */
class MyMapFunction2 extends MapFunction[Int, Int] {

  override def map(value: Int) = value * 2

}
